package cn.edu.flink.scala.tutorial.cep

import org.apache.flink.cep.functions.{PatternProcessFunction, TimedOutPartialMatchHandler}
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector
import java.util

object TimedOutTest {
  case class OrderEvent(userId: String, orderId: String, eventType: String, timestamp: Long)

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    val orderStream = env.fromElements(
      OrderEvent("user_1", "order_1", "create", 1000L),
      OrderEvent("user_2", "order_2", "create", 2000L),
      OrderEvent("user_1", "order_1", "modify", 3000L),
      OrderEvent("user_1", "order_1", "pay", 24000L),
      OrderEvent("user_2", "order_3", "create", 25000L),
      OrderEvent("user_2", "order_3", "pay", 26000L))
      .assignAscendingTimestamps(_.timestamp)
      .keyBy(_.orderId)

    val pattern = Pattern
      .begin[OrderEvent]("create")
      .where(_.eventType.equals("create"))
      .followedBy("pay")
      .where(_.eventType.equals("pay"))
      .within(Time.seconds(10))

    val patternStream = CEP.pattern(orderStream, pattern)

    val timeOutTag = new OutputTag[String]("timeout")
    val payedStream = patternStream.process(new OredrPayPatternProcessFunction(timeOutTag))

    payedStream.print("payed")
    payedStream.getSideOutput(timeOutTag).print("timeout")

    env.execute("TimedOutTest")
  }

  class OredrPayPatternProcessFunction(timeOutTag: OutputTag[String]) extends PatternProcessFunction[OrderEvent, String] with TimedOutPartialMatchHandler[OrderEvent] {
    override def processMatch(map: util.Map[String, util.List[OrderEvent]], context: PatternProcessFunction.Context, collector: Collector[String]): Unit = {
      val payEvent = map.get("pay").get(0)
      collector.collect(s"订单 ${payEvent.orderId} 已支付")
    }


    override def processTimedOutMatch(map: util.Map[String, util.List[OrderEvent]], context: PatternProcessFunction.Context): Unit = {
      val createOrder = map.get("create").get(0)

      context.output(timeOutTag, s"订单 ${createOrder.orderId} 超时未支付！ 用户为: ${createOrder.userId}")
    }
  }
}
